package com.shujia.mapdesk;

import java.io.*;
import java.util.HashMap;

public class MapDesk implements Runnable{
    private File block;
    private String blockName;
    BufferedReader br = null;
    BufferedWriter bw = null;

    public MapDesk(File block, String blockName) {
        this.block = block;
        this.blockName = blockName;


        try {
            br = new BufferedReader(new FileReader(block.getAbsoluteFile()));
            bw = new BufferedWriter(new FileWriter("hodoop/out/part-m-0000"+blockName.substring(5)));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    @Override
    public void run() {
        HashMap<String, Long> map = new HashMap<>();
        String line = null;
        try {
            while ((line= br.readLine())!=null){
                String clazz = line.split(",")[4];
                if (map.containsKey(clazz)) {
                    map.put(clazz,map.get(clazz)+1);
                }else {
                    map.put(clazz,1L);
                }
            }
            map.forEach((k, v) -> {
                try {
                    bw.write(k+" "+v);
                    bw.newLine();
                    bw.flush();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            });
        }catch (Exception e){
            e.printStackTrace();
        }
        try {
            br.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        try {
            bw.close();
        } catch (IOException e) {
            e.printStackTrace();
        }

    }
}
